package org.mq.mymq.nameserver.replication;

import com.alibaba.fastjson.JSON;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import org.mq.mymq.common.coder.TcpMsg;
import org.mq.mymq.common.dto.SlaveAckDTO;
import org.mq.mymq.common.enums.NameServerEventCode;
import org.mq.mymq.common.enums.NameServerResponseCode;
import org.mq.mymq.nameserver.common.CommonCache;
import org.mq.mymq.nameserver.common.MasterSlaveReplicationProperties;
import org.mq.mymq.nameserver.enums.MasterSlaveReplicationTypeEnum;
import org.mq.mymq.nameserver.event.model.ReplicationMsgEvent;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @ClassName MasterSlaveReplicationMsgSendTask
 * @Description 主从同步专用的数据发送任务
 * @Author jiarong_ye
 * @Date 2025/6/20 12:45
 * @Version 1.0
 */
public class MasterReplicationMsgSendTask extends ReplicationTask{

    public MasterReplicationMsgSendTask(String taskName) {
        super(taskName);
    }

    @Override
    public void startTask() {
        MasterSlaveReplicationProperties masterSlaveReplicationProperties = CommonCache.getNameserverProperties().getMasterSlaveReplicationProperties();
        // 判断当前复制模式，如果是异步复制，直接发送同步数据，同时返回注册成功信号给到broker节点
        // 如果是同步复制，发送同步数据给到slave节点，slave节点返回ack信号，在主节点收到ack信号后通知broker注册成功
        // 版同步复制和同步复制相似
        MasterSlaveReplicationTypeEnum masterSlaveReplicationTypeEnum = MasterSlaveReplicationTypeEnum.of(masterSlaveReplicationProperties.getType());
        while (true) {
            try {
                ReplicationMsgEvent replicationMsgEvent = CommonCache.getReplicationMsgQueueManager().getReplicationMsgQueue().take();
                replicationMsgEvent.setMsgId(UUID.randomUUID().toString());
                Channel brokerChannel = replicationMsgEvent.getChannelHandlerContext().channel();
                Map<String, ChannelHandlerContext> channelHandlerContextMap = CommonCache.getReplicationChannelManager().getValidSlaveChannelMap();
                int validSlaveChannelCount = channelHandlerContextMap.keySet().size();
                // 给从节点发送同步数据 并返回ack
                // broker发送心跳数据到nameserver->nameserver获取
                // sync 同步复制 每条消息同步发送给所有的从节点
                // 半同步复制
                // async 异步复制
                if (masterSlaveReplicationTypeEnum == MasterSlaveReplicationTypeEnum.ASYNC) {
                    this.sendMsgToSlave(replicationMsgEvent);
                    brokerChannel.writeAndFlush(new TcpMsg(NameServerResponseCode.REGISTRY_SUCCESS.getCode(), NameServerResponseCode.REGISTRY_SUCCESS.getDesc().getBytes()));
                } else if (masterSlaveReplicationTypeEnum == MasterSlaveReplicationTypeEnum.SYNC) {
                    this.inputMsgToAckMap(replicationMsgEvent, validSlaveChannelCount);
                    this.sendMsgToSlave(replicationMsgEvent);
                } else if (masterSlaveReplicationTypeEnum == MasterSlaveReplicationTypeEnum.HALF_SYNC) {
                    this.inputMsgToAckMap(replicationMsgEvent, validSlaveChannelCount / 2);
                    this.sendMsgToSlave(replicationMsgEvent);
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }

    /**
     * 将主节点需要同步的消息和消息需要ack次数保存到map中
     *
     * @param replicationMsgEvent
     * @param needAckCount
     */
    private void inputMsgToAckMap(ReplicationMsgEvent replicationMsgEvent, int needAckCount) {
        CommonCache.getAckMap().put(replicationMsgEvent.getMsgId(),
                new SlaveAckDTO(new AtomicInteger(needAckCount),
                        replicationMsgEvent.getChannelHandlerContext()));
    }

    /**
     * 发送数据到从节点
     *
     * @param replicationMsgEvent
     */
    private void sendMsgToSlave(ReplicationMsgEvent replicationMsgEvent) {
        Map<String, ChannelHandlerContext> channelHandlerContextMap = CommonCache.getReplicationChannelManager().getValidSlaveChannelMap();
        List<String> inValidChannelReqIdList = new ArrayList<>();
        for (String reqId : channelHandlerContextMap.keySet()) {
            ChannelHandlerContext ctx = channelHandlerContextMap.get(reqId);
            replicationMsgEvent.setChannelHandlerContext(null);
            byte[] body = JSON.toJSONBytes(replicationMsgEvent);
            ctx.writeAndFlush(new TcpMsg(NameServerEventCode.MASTER_REPLICATION_MSG.getCode(), body));
        }
    }
}
